#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <assert.h>

#include "base.h"
#include "thread.h"

const char * TH_TAG[] = {
						"[worker]", 
						"[master]", 
						"[logger]", 
						"[timer ]", 
						};

/* Number of threads that have finished setting themselves up */
static int threads_count = 0; /* count of running threads */
static pthread_mutex_t threads_lock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t threads_cond = PTHREAD_COND_INITIALIZER;

void threads_wait(int count)
{
	pthread_mutex_lock(&threads_lock);
    while (threads_count != count) {
        pthread_cond_wait(&threads_cond, &threads_lock);
    }
    pthread_mutex_unlock(&threads_lock);
	return;
}

static void cond_signal(int val)
{
	pthread_mutex_lock(&threads_lock);
    threads_count += val;
    pthread_cond_signal(&threads_cond);
    pthread_mutex_unlock(&threads_lock);
	return;
}

static __inline__ void inc_signal() { return cond_signal(+1); }
static __inline__ void dec_signal() { return cond_signal(-1); }

/* -- send request to worker -- */
int send_request(TASK * task, THREAD * p)
{
	assert(task && p);

	pthread_mutex_lock(&p->request_lock);
	//write new task
	list_add_tail(&task->list, &p->request_q);
	//write pipe
	if (write(p->request_send_fd, "", 1) != 1) {
		send_log(LV_ERROR, "#send_request# write pipe %d : %s",
					p->request_send_fd, strerror(errno));
		return -1;
	}
	pthread_mutex_unlock(&p->request_lock);
	return 0;
}

/* -- worker recv request -- */
TASK * recv_request(THREAD * p)
{
	assert(p);

	char buf[1];

	pthread_mutex_lock(&p->request_lock);
	//read pipe
	if (read(p->request_recv_fd, buf, 1) != 1){
		send_log(LV_ERROR, "#recv_request# read pipe %d : %s", 
					p->request_recv_fd, strerror(errno));
		return NULL;
	}
	//read new task
	if (list_empty(&p->request_q)) {
		send_log(LV_ERROR, "#recv_request# list empty");
		return NULL;
	}
	TASK * task = list_entry(p->request_q.next, TASK, list);
	assert(task);
	list_del(&task->list);
	pthread_mutex_unlock(&p->request_lock);
	return task;
}

/* -- send response to master -- */
int send_response(TASK * task, THREAD * p)
{
	assert(task && p);

	pthread_mutex_lock(&p->response_lock);
	//write new task
	list_add_tail(&task->list, &p->response_q);
	//write pipe
	if (write(p->response_send_fd, "", 1) != 1) {
		send_log(LV_ERROR, "#send_response# write pipe %d : %s", 
					p->response_send_fd, strerror(errno));
		return -1;
	}
	pthread_mutex_unlock(&p->response_lock);
	return 0;
}

/* -- master recv response -- */
TASK * recv_response(THREAD * p)
{
	assert(p);

	char buf[1];

	pthread_mutex_lock(&p->response_lock);
	//read pipe
	if (read(p->response_recv_fd, buf, 1) != 1){
		send_log(LV_ERROR, "#recv_response# read pipe %d : %s", 
					p->response_recv_fd, strerror(errno));
		return NULL;
	}
	//read new task
	if (list_empty(&p->response_q)) {
		send_log(LV_ERROR, "#recv_response# list empty");
		return NULL;
	}
	TASK * task = list_entry(p->response_q.next, TASK, list);
	assert(task);
	list_del(&task->list);
	pthread_mutex_unlock(&p->response_lock);
	return task;
}

//bind with thread request event
static void thread_notify_handler(int fd, short which, void *arg)
{
	assert(fd && arg);

	THREAD * p = (THREAD *)arg;
	TASK * task = recv_request(p);
	if (!task) {
		return;
	}
	if (p->ontask(task, p) < 0) {
		return;
	}
	return;
}

int thread_free(THREAD * p)
{
	if (!p) return 1;

	event_del(&p->request_event);

	if (p->base) {
		event_base_free(p->base);
		p->base = NULL;
	}
	//free request_q
	pthread_mutex_destroy(&p->request_lock);
	pthread_mutex_destroy(&p->response_lock);
	free(p);
	return 0;
}

THREAD * thread_new(TH_ONTASK_FUNC ontask, TH_INIT_FUNC init,
						TH_CLEAR_FUNC clear, void * sys)
{
	THREAD * p = malloc(sizeof(struct THREAD));
	if (p == NULL) {
		return NULL;
	}

	p->tid = pthread_self();
	p->type = TYPE_MASTER;
	p->ontask = ontask;
	p->init = init;
	p->clear = clear;
	p->sys = sys;

	p->base = event_base_new();
	if(p->base == NULL) {
		thread_free(p);
		send_log(LV_SYS, "allocate event base failed");
		return NULL;
	}

	int fds[2];
	if (pipe(fds)) {
		thread_free(p);
		send_log(LV_SYS, "create notify pipe failed: %s", strerror(errno));
		return NULL;
	}
	p->request_recv_fd = fds[0];
	p->request_send_fd = fds[1];
	if (pipe(fds)) {
		thread_free(p);
		send_log(LV_SYS, "create notify pipe failed: %s", strerror(errno));
		return NULL;
	}
	p->response_recv_fd = fds[0];
	p->response_send_fd = fds[1];

	/* Listen for notifications from other threads */
	event_set(&p->request_event, p->request_recv_fd,
			EV_READ | EV_PERSIST, thread_notify_handler, p);
	event_base_set(p->base, &p->request_event);
	if (event_add(&p->request_event, 0) == -1) {
		thread_free(p);
		send_log(LV_SYS, "monitor libevent notify pipe failed");
		return NULL;
	}
	INIT_LIST_HEAD(&p->request_q);
	INIT_LIST_HEAD(&p->response_q);
	pthread_mutex_init(&p->request_lock, NULL);
	pthread_mutex_init(&p->response_lock, NULL);

	return p;
}

static void * thread_routine(void *arg)
{
	THREAD * p = (THREAD *)arg;

	p->type = TYPE_WORKER;
	if (threads_count == 0)
		p->type = TYPE_LOGGER;
	else if (!p->ontask)
		p->type = TYPE_TIMER;

	if (p->init && p->init(p) < 0) {
		send_log(LV_ERROR, "%s init", TTAG(p));
		return NULL;
	}

	send_log(LV_SYS, "%s request[r(%d)s(%d)] response[r(%d)s(%d)] start up", 
				TTAG(p), p->request_recv_fd, p->request_send_fd, 
				p->response_recv_fd, p->response_send_fd);

	inc_signal();
	p->running = 1;

	event_base_loop(p->base, 0);

	return NULL;
}

int thread_startup(THREAD * p)
{
	pthread_attr_t	attr;
	int				ret;

	pthread_attr_init(&attr);

	if ((ret = pthread_create(&p->tid, &attr, thread_routine, (void*)p)) != 0) {
			send_log(LV_SYS, "thread create failed, %s, exit", strerror(errno));
		return -1;
	}
	return 0;
}

int thread_shutdown(THREAD * p)
{
	if (!p->running) return 1;

	p->running = 0;

	if (p->clear && p->clear(p) < 0) {
		send_log(LV_ERROR, "%s clear", TTAG(p));
	}

	send_log(LV_SYS, "%s request[r(%d)s(%d)] response[r(%d)s(%d)] shut down", 
				TTAG(p), p->request_recv_fd, p->request_send_fd, 
				p->response_recv_fd, p->response_send_fd);

	dec_signal();

	pthread_cancel(p->tid);
	return 0;
}
